package com.czh.dubbo.consumer.filter;

import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.BinaryOperator;
import java.util.stream.Collectors;

@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter implements Filter {

    private static class Record {
        String methodName;
        long cost;
        long endTime;

        public Record(String methodName, long cost, long endTime) {
            this.methodName = methodName;
            this.cost = cost;
            this.endTime = endTime;
        }
    }

    public TPMonitorFilter() {
        Executors.newScheduledThreadPool(1).scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                queue.drainTo(recordList);

                long current = System.currentTimeMillis();
                recordList = recordList.stream()
                        .filter(record -> current - record.endTime <= 60000)
                        .collect(Collectors.toCollection(LinkedList::new));

                recordList.stream()
                        .collect(Collectors.toMap(record -> record.methodName, record -> {
                            LinkedList<Record> result = new LinkedList<>();
                            result.add(record);
                            return result;
                        }, (BinaryOperator<List<Record>>) (records, records2) -> {
                            records.addAll(records2);
                            return records;
                        }))
                        .forEach((s, records) -> {
                            records.sort(Comparator.comparingLong(o -> o.cost));
                            System.out.printf("method:%s,tp90:%dms,tp99:%dms\n", s,
                                    records.get(((int) (records.size() * 0.9d))).cost,
                                    records.get(((int) (records.size() * 0.99d))).cost);
                        });
            }
        }, 5, 5, TimeUnit.SECONDS);
    }

    List<Record> recordList = new LinkedList<>();

    LinkedBlockingQueue<Record> queue = new LinkedBlockingQueue<>();

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        long startTime = System.currentTimeMillis();
        Result result = invoker.invoke(invocation);
        long endTime = System.currentTimeMillis();
        queue.add(new Record(invocation.getMethodName(), endTime - startTime, endTime));
        return result;
    }
}
